-- 数据库服务
-- Author: Colocaisa
-- Date: 2018-03-14

local skynet = require "skynet";
require "skynet.manager";
local log = require "log";
local dbstruct = require "common.db.dbstruct";
local dbdefine = require "common.db.dbdefine";
local string = require "us.string";
local table = table;
local setmetatable = table.setmetatable;

local us_db = {}

local mysql_pool = nil;
local redis_pool = nil;
local us_dbsync = nil;

local dbname = "test";
local dbtbs = {};
local tbfields = {};
local tbpks = {};

-- redis更新列表
local ulorder = 0;
local ulkey = "update_list";


-- 版本控制
local rdver = {};

local ENUM_FIELDS_TYPE_DEFAULT = 
{
    ['tinyint']     = function() return 0   end,
    ['smallint']    = function() return 0   end,
    ['mediumint']   = function() return 0   end,
    ['int']         = function() return 0   end,
    ['bigint']      = function() return 0   end,
    ['float']       = function() return 0.0 end,
    ['double']      = function() return 0.0 end,
    ['decimal']     = function() return 0.0 end,
    ['char']        = function() return ''  end,
    ['varchar']     = function() return ''  end,
    ['tinyblob']    = function() return ''  end,
    ['tinytext']    = function() return ''  end,
    ['blob']        = function() return ''  end,
    ['text']        = function() return ''  end,
    ['mediumblob']  = function() return ''  end,
    ['mediumtext']  = function() return ''  end,
    ['longblob']    = function() return ''  end,
    ['longtext']    = function() return ''  end,


    ['date']        = function() 
                          return os.date("%Y%m%d", os.time());
                      end,

    ['time']        = function() 
                          return os.date("%H%M%S", os.time());
                      end,

    ['year']        = function() 
                          return os.date("%Y", os.time());
                      end,

    ['datetime']    = function() 
                          return os.date("%Y%m%d%H%M%S", os.time());
                      end,

    ['timestamp']   = function() 
                          return os.date("%Y%m%d%H%M%S", os.time());
                      end,
}


local ENUM_FIELDS_TYPE_CONVERT = 
{
    ['tinyint']     = function(v) return tonumber(v) end,
    ['smallint']    = function(v) return tonumber(v) end,
    ['mediumint']   = function(v) return tonumber(v) end,
    ['int']         = function(v) return tonumber(v) end,
    ['bigint']      = function(v) return tonumber(v) end,
    ['float']       = function(v) return tonumber(v) end,
    ['double']      = function(v) return tonumber(v) end,
    ['decimal']     = function(v) return tonumber(v) end,
    ['char']        = function(v) return v end,
    ['varchar']     = function(v) return v end,
    ['tinyblob']    = function(v) return v end,
    ['tinytext']    = function(v) return v end,
    ['blob']        = function(v) return v end,
    ['text']        = function(v) return v end,
    ['mediumblob']  = function(v) return v end,
    ['mediumtext']  = function(v) return v end,
    ['longblob']    = function(v) return v end,
    ['longtext']    = function(v) return v end,
    ['date']        = function(v) return v end,
    ['time']        = function(v) return v end,
    ['year']        = function(v) return v end,
    ['datetime']    = function(v) return v end,
    ['timestamp']   = function(v) return v end,
}


local function get_db_tables()
    local sql = string.format("CALL get_db_tables(\'%s\');", dbname);
    local rs = skynet.call(mysql_pool, "lua", "execute", sql);
    local rt = {}
    for k, v in pairs(rs[1]) do
        table.insert(rt, v["TABLE_NAME"]);
    end
    return rt;
end

local function get_primary_key(tbname)
    local sql = string.format("CALL get_primary_key(\'%s\', \'%s\');", dbname, tbname);
    local rs = skynet.call(mysql_pool, "lua", "execute", sql);
    return rs[1][1]["COLUMN_NAME"];
end

local function get_fields(tbname)
    local sql = string.format("CALL get_fields(\'%s\', \'%s\');", dbname, tbname);
    local rs = skynet.call(mysql_pool, "lua", "execute", sql);
    local fields = {};
    for _, row in pairs(rs[1]) do
        table.insert(fields, row["column_name"]);
    end
    return fields;
end

local function get_field_type(tbname, field)
    local sql = string.format("CALL get_field_type(\'%s\', \'%s\', \'%s\');", dbname, tbname, field);
    local rs = skynet.call(mysql_pool, "lua", "execute", sql);
    return rs[1][1]["data_type"];
end

local function get_auto_increment(tbname)
    local sql = string.format("CALL get_auto_increment(\'%s\', \'%s\');", dbname, tbname);
    local rs = skynet.call(mysql_pool, "lua", "execute", sql);
    return rs[1][1]["AUTO_INCREMENT"];
end

local function _get_tab_key_equal_value(tk, tv)
    local ret = {};
    assert(#tk == #tv);
    local cnt = #tv;
    for i=1, cnt do
        ret[tk[i]] = tv[i];
    end
    return ret;
end

local function _get_tab_key_value(tk, tv)
    local ret = {}
    assert(#tk == #tv);
    local cnt = #tv;
    for i=1, cnt do
        table.insert(ret, tk[i]);
        table.insert(ret, tv[i]);
    end
    return ret;
end

local function _get_tab_keys_tab_values(t)
    local rk = {}
    local rv = {}
    for k, v in pairs(t) do
        table.insert(rk, k);
        table.insert(rv, v);
    end
    return rk, rv
end

local function _get_str_key_equal_value(t)
    local ret = {};
    for k, v in pairs(t) do
        table.insert(ret, k .. " = \'" .. v .. "\'")
    end
    return ret;
end

local function _pre_handle_cmd(str)
    return string.gsub(str, "\'%$(%a+%(%))\'", "%1");
end

local function _get_table_fields(tbname)
    if nil ~= tbfields[tbname] then
        return tbfields[tbname];
    end

    tbfields[tbname] = {};
    local pfields = tbfields[tbname];

    local fields = get_fields(tbname);
    for k, v in pairs(fields) do
        pfields[v] = get_field_type(tbname, v);
    end

    return pfields;
end

local function _get_table_pk(tbname)
    if nil == tbpks[tbname] then
        tbpks[tbname] = get_primary_key(tbname);
    end
    return tbpks[tbname];
end

local function _get_local_ver(tbname, tbpk)
    if nil == rdver[tbname] then
        rdver[tbname] = {};
    end
    local ptbver = rdver[tbname];
    if nil == ptbver[tbpk] then
        ptbver[tbpk] = 0;
    end

    local rt = ptbver[tbpk];
    if rt >= 0 then
        ptbver[tbpk] = rt + 1;
    end

    return rt
end


local function _check_rd_ver(tbname, tbpk, ver)
    assert(ver, "ver can not be null");
    if nil == rdver[tbname] then
        rdver[tbname] = {};
    end
    local ptbver = rdver[tbname];
    if nil == ptbver[tbpk] then
        ptbver[tbpk] = 0;
    end
    local rt = true;

    local orverkey = tbname.."_ver:"..tbpk
    local orver = skynet.call(redis_pool, "lua", "get", tbpk, orverkey);

    if ver < 0 or nil == orver or (orver >= 0 and orver < ver) then
        rt = true;
    else
        rt = false;
    end

    if true == rt then
        ver = ver or -1;
        ptbver[tbpk] = ver;
        skynet.call(redis_pool, "lua", "set", tbpk, orverkey, ver);
        log.printf("table %s primary key %s record to redis ok version is %s", tbname, tbpk, ver);
    end
    log.printf("table %s key %s origin ver ==> %s", tbname, tbpk, orver);
    log.printf("table %s key %s local ver ==> %s", tbname, tbpk, ver);

    return rt
end

local function _init_tbcfg_to_redis(tbname)
    if nil ~= dbtbs[tbname] then
        return dbtbs[tbname];
    end

    local r = true;
    local tbpkf = _get_table_pk(tbname);
    log.dump(tbpkf);
    local incr = get_auto_increment(tbname);
    local k = tbname..":"..tbpkf;
    local rs = skynet.call(redis_pool, "lua", "set", nil, k, incr-1);

    log.printf("init %s to redis %s ==> %s, result ==> %s", tbname, k, incr, rs);

    if true == r then
        dbtbs[tbname] = true;
    end
    return r
end

local function _load_data_to_redis()
    local tbs = get_db_tables();

    for k, tbname in pairs(tbs) do
        _init_tbcfg_to_redis(tbname);
    end
end

local function _dbc_to_rdc(dbc, c)
    local pk = assert(dbc.m_pk, "convert to rdc must set m_pk field!")
    local rdc = {}
    table.insert(rdc, c);

    local k = dbc.m_tbname..":"..pk;
    table.insert(rdc, k);

    local v = _get_tab_key_value(dbc.m_columns, dbc.m_values);
    table.insert(rdc, v);

    return rdc;
end

local function _rdr_to_dbr(dbc, rdr)
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local pfields = _get_table_fields(tbname);

    local rt = nil;
    local cnt = #rdr;
    if cnt > 0 then
        rt = {};
    end

    for i=1, cnt, 2 do
        local f = rdr[i];
        local v = rdr[i+1];
        local ft = pfields[f];
        local c = ft and ENUM_FIELDS_TYPE_CONVERT[ft];
        rt[f] = c and c(v) or v;
        -- log.printf("field ==> %s    type ==> %s    value ==> %s", f, type(rt[f]), rt[f]);
    end

    return rt
end

function us_db.test()
    --skynet.call(mysql_pool, "lua", "test");
    local dbc = dbstruct.create_cmd("account");
    -- dbc.m_schema = { 
    --     acc_name = "woca122",
    --     acc_pwd = "123456",
    -- }

    dbc.m_columns = {"acc_name", "acc_pwd"};
    dbc.m_values = {"woca1004", "123456"};
    dbc.m_condition = {};

    local res = us_db.insert(dbc, true);
    -- log.dump(res);

    us_db.update(dbc);

    res = us_db.delete(dbc);
    -- log.dump(res);
    --[[
    local dbcs = dbstruct.create_cmd("account");
    dbcs.m_pk = 10000001;
    dbcs.m_columns = {"acc_name", "acc_pwd"};
    dbcs.m_values = {"woca1001", "11321321312"};
    us_db.update(dbcs, true);

    local dbcs = dbstruct.create_cmd("account");
    dbcs.m_pk = 10000001;
    local rs = us_db.select(dbcs);
    log.dump(rs, "result", 6);
    ]]
    -- skynet.call(mysql_pool, "lua", "test", dbc);
end


local function _handle_result(res, dbc)
    local success = 0;
    local failed = 0;
    local errcode;
    local errmsg;

    local function __handle_result(result)
        if true == result.badresult then
            log.printf("database operation failed errcode %s: %s", result.errno, result.err);
            failed = failed + 1;
            errcode = errcode or result.errno;
            errmsg = errmsg or result.err;
        else
            local __id = result.insert_id;
            if nil ~= __id and nil == dbc.m_pk then
                dbc.m_pk = __id;
            end
            success = success + 1;
        end
    end

    local bmulit = res.mulitresultset;
    if true == bmulit then
        for _, result in ipairs(res) do
            __handle_result(result);
        end
    else
        __handle_result(res);
        res = {res};
    end

    res.success = success;
    res.failed = failed;
    res.errcode = errcode;
    res.errmsg = errmsg;

    return res;
end


-- 插入
-------------------------------------------------------------------
-- 初始化一行数据
local function _init_dbc(dbc)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");

    local tbschema = _get_tab_key_equal_value(dbc.m_columns, dbc.m_values);
    local pfields = _get_table_fields(tbname);
    local tbpkf = _get_table_pk(tbname);

    if nil == dbc.m_pk then
        dbc.m_pk = skynet.call(redis_pool, "lua", "incr", nil, tbname..":"..tbpkf);
        log.printf("get pk from redis %s ==> %s  type ==> %s", tbpkf, dbc.m_pk, type(dbc.m_pk));
    end
    local tbpkv = dbc.m_pk;

    -- 如果主键字段为空
    tbschema[tbpkf] = tbschema[tbpkf] or tbpkv;
    -- 没有的字段填充初始值
    for f, ft in pairs(pfields) do
        tbschema[f] = tbschema[f] or ENUM_FIELDS_TYPE_DEFAULT[ft]()
    end

    local tk, tv = _get_tab_keys_tab_values(tbschema);
    dbc.m_columns = tk;
    dbc.m_values = tv;
end


-- 插入数据到mysql
local function _insert_mysql(dbc)
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local str = ""
    if 1 == dbc.m_exetype then           -- SQL
        local p_tcolumns = dbc.m_columns;
        local p_tvalues = dbc.m_values;
        str = "INSERT INTO " .. tbname .." ";
        str = str .. "( " .. table.concat(p_tcolumns, ", ") .. " ) ";
        str = str .. "VALUES ";
        str = str .. "( \'" .. table.concat(p_tvalues, "\', \'") .. "\');";
    elseif 2 == dbc.m_exetype then       -- PROCEDURE
        str = "CALL " .. dbc.m_execmd .. " ( \'" .. table.concat(dbc.m_args, "\', \'") .. "\' );";
    end
    log.printf(str);
    return _handle_result(skynet.call(mysql_pool, "lua", "execute", _pre_handle_cmd(str), true), dbc);
end

-- 插入数据到redis
local function _insert_redis(dbc, ver);
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpk = assert(dbc.m_pk, "db cmd m_pk must be not null");

    local sync = _check_rd_ver(tbname, tbpk, ver);
    if false == sync then
        return false;
    end

    local rdc = _dbc_to_rdc(dbc, "hmset");
    local rs = skynet.call(redis_pool, "lua", table.remove(rdc, 1), dbc.conn, table.unpack(rdc));
    log.printf("insert to redis pk is %s", tbpk);
    _check_rd_ver(tbname, tbpk, -1);

    return true;
end

function us_db.insert(dbc, sync)
    assert(dbc, "db cmd must be not null");
    local code = dbdefine.ENUM_DBRESULT_ERRCODE.SUCCESS;
    local derr;
    local dmsg;
    local rs;

    if true == sync then
        rs = _insert_mysql(dbc);
    end

    if nil ~= rs and rs.failed > 0 then
        code = dbdefine.ENUM_DBRESULT_ERRCODE.DBFAILED;
        derr = rs.errcode;
        dmsg = rs.errmsg;
    end

    if dbdefine.ENUM_DBRESULT_ERRCODE.SUCCESS == code then
        _init_dbc(dbc);
        log.dump(dbc);

        rs = _insert_redis(dbc, -1);
        log.dump(rs);
    end

    if nil == rs then
        code = dbdefine.ENUM_DBRESULT_ERRCODE.RDFAILED;
    end

    return dbstruct.create_result(code, derr, dmsg);
end
-------------------------------------------------------------------



-- 查询
-------------------------------------------------------------------
-- 查询redis
local function _select_redis(dbc)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpkv = assert(dbc.m_pk, "db cmd m_pk must be not null");

    local rekey = tbname..":"..tbpkv;
    local rs = skynet.call(redis_pool, "lua", "hgetall", tbpkv, rekey);

    return _rdr_to_dbr(dbc, rs);
end

-- 查询mysql
function _select_mysql(dbc)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpkv = assert(dbc.m_pk, "db cmd m_pk must be not null");
    local rt = nil;

    local str = ""
    if 1 == dbc.m_exetype then           -- SQL
        local tbcond = dbc.m_condition;
        local tbpkf = _get_table_pk(tbname);
        tbcond[tbpkf] = tbcond[tbpkf] or tbpkv;
        local p_tcolumns = dbc.m_columns;
        if #p_tcolumns == 0 then 
            table.insert(p_tcolumns, "*");
        end
        local condtab = _get_str_key_equal_value(tbcond);
        str = "SELECT " .. table.concat(p_tcolumns, ", ");
        str = str .. " FROM " .. tbname;
        str = str .. " WHERE ";
        str = str .. table.concat(condtab, " AND ") .. ";";
    elseif 2 == dbc.m_exetype then       -- PROCEDURE
        str = "CALL " .. dbc.m_execmd .. " ( \'" .. table.concat(dbc.m_args, "\', \'") .. "\' );";
    end
    log.printf(str)
    rt = _handle_result(skynet.call(mysql_pool, "lua", "execute", _pre_handle_cmd(str)), dbc);

    return rt;
end

-- 用结果初始化dbc
local function _init_dbc_by_rd(dbc, rd)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpkv = assert(dbc.m_pk, "db cmd m_pk must be not null");
    local tbpkf = _get_table_pk(tbname);

    if tbpkv ~= rd[tbpkf] then
        return false;
    end

    local tk, tv = _get_tab_keys_tab_values(rd);
    dbc.m_columns = tk;
    dbc.m_values = tv;

    log.dump(dbc);
    return true;
end

-- 同步记录到redis
local function _sync_to_redis(dbc, rt, ver)
    assert(dbc, "db cmd must be not null");
    assert(rt, "rt must be not null");

    local cnt = #rt;
    for i=1, cnt do
        local records = rt[i];
        for _, record in pairs(records) do
            if true == _init_dbc_by_rd(dbc, record) then
                return _insert_redis(dbc, ver);
            end
        end
    end
    return false;
end


function us_db.select(dbc)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpkv = assert(dbc.m_pk, "db cmd m_pk must be not null");
    local code = dbdefine.ENUM_DBRESULT_ERRCODE.SUCCESS;
    local derr;
    local dmsg;

    local rt = _select_redis(dbc);
    if rt ~= nil then
        return dbstruct.create_result(code, nil, rt);
    end

    local ver = _get_local_ver(tbname, tbpkv);

    rt = _select_mysql(dbc);
    local cnt = #rt;
    if rt.failed > 0 or 0 == cnt then
        code = dbdefine.ENUM_DBRESULT_ERRCODE.DBFAILED;
        derr = rt.errcode;
        dmsg = rt.errmsg;
        return dbstruct.create_result(code, derr, dmsg);
    end

    log.dump(rt);

    if false == _sync_to_redis(dbc, rt, ver) then
        code = dbdefine.ENUM_DBRESULT_ERRCODE.RDFAILED;
        return dbstruct.create_result(code, nil, rt);
    end

    -- log.dump(rt);
    local tbschema = _get_tab_key_equal_value(dbc.m_columns, dbc.m_values);

    return dbstruct.create_result(code, nil, tbschema);
end

-------------------------------------------------------------------


-- 删除
-------------------------------------------------------------------
-- 删除redis
local function _delete_redis(dbc)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpkv = assert(dbc.m_pk, "db cmd m_pk must be not null");

    local rekey = tbname..":"..tbpkv;
    local rs = skynet.call(redis_pool, "lua", "hset", dbc.conn or tbpkv, rekey, "_del", "true");
    return true;
end

-- 移除记录
local function _delete_cache(dbc)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpkv = assert(dbc.m_pk, "db cmd m_pk must be not null");

    local rekey = tbname..":"..tbpkv;
    local rs = skynet.call(redis_pool, "lua", "del", dbc.conn or tbpkv, rekey);
    return true;
end

-- 删除mysql
local function _delete_mysql(dbc)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpkv = assert(dbc.m_pk, "db cmd m_pk must be not null");
    local tbpkf = _get_table_pk(tbname);
    local tbcond = dbc.m_condition;
    tbcond[tbpkf] = tbcond[tbpkf] or tbpkv;

    local str = "";
    if 1 == dbc.m_exetype then 
        str = "DELETE FROM " .. dbc.m_tbname.." ";
        str = str .. "WHERE ";
        local condtab = _get_str_key_equal_value(tbcond);
        str = str .. table.concat(condtab, " AND ") .. ";";
    else 
        str = "CALL " .. dbc.m_execmd .. " ( \'" .. table.concat(dbc.m_args, "\', \'") .. "\' );";
    end
    log.printf("execute SQL==>%s", _pre_handle_cmd(str));
    return _handle_result(skynet.call(mysql_pool, "lua", "execute", _pre_handle_cmd(str), true), dbc);
end


function us_db.delete(dbc, sync)
    assert(dbc, "db cmd must be not null");
    local code = dbdefine.ENUM_DBRESULT_ERRCODE.SUCCESS;
    local derr;
    local dmsg;

    if false == _delete_redis(dbc) then
        code = dbdefine.ENUM_DBRESULT_ERRCODE.RDFAILED;
        return dbstruct.create_result(code, derr, dmsg);
    end

    local rs = _delete_mysql(dbc);
    if nil ~= rs and rs.failed > 0 then
        code = dbdefine.ENUM_DBRESULT_ERRCODE.DBFAILED;
        derr = rs.errcode;
        dmsg = rs.errmsg;
        return dbstruct.create_result(code, derr, dmsg);
    end

    if true == sync and false == _delete_cache(dbc) then
        code = dbdefine.ENUM_DBRESULT_ERRCODE.RDFAILED;
    end
    return dbstruct.create_result(code, derr, dmsg);
end
-------------------------------------------------------------------


-- 更新
-------------------------------------------------------------------
local function _update_redis(dbc)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpkv = assert(dbc.m_pk, "db cmd m_pk must be not null");

    local rdc = _dbc_to_rdc(dbc, "hmset");
    local rs = skynet.call(redis_pool, "lua", table.remove(rdc, 1), dbc.conn or tbpkv, table.unpack(rdc));
    if rs == "OK" then
        local rediskey = tbname..":"..tbpkv;
        ulorder = ulorder + 1;
        skynet.send(redis_pool, "lua", "zadd", nil, ulkey, ulorder, rediskey);
        return true;
    else
        return false;
    end
end


local function _update_mysql(dbc)
    assert(dbc, "db cmd must be not null");
    local tbname = assert(dbc.m_tbname, "db cmd m_tbname must be not null");
    local tbpkv = assert(dbc.m_pk, "db cmd m_pk must be not null");
    local tbpkf = _get_table_pk(tbname);
    local tbcond = dbc.m_condition;
    tbcond[tbpkf] = tbcond[tbpkf] or tbpkv;

    -- 更新mysql
    local str = "";
    if 1 == dbc.m_exetype then 
        str = "UPDATE " .. tbname.." SET ";
        local tbschema = _get_tab_key_equal_value(dbc.m_columns, dbc.m_values);
        local strtab = _get_str_key_equal_value(tbschema);
        str = str .. table.concat(strtab, ", ") .. " ";
        local condtab = _get_str_key_equal_value(tbcond);
        str = str .. "WHERE ";
        str = str .. table.concat(condtab, " AND ") .. ";";
    else 
        str = "CALL " .. dbc.m_execmd .. " ( \'" .. table.concat(dbc.m_args, "\', \'") .. "\' );";
    end

    log.printf("execute SQL==>%s", _pre_handle_cmd(str));
    skynet.send(us_dbsync, "lua", "sync", _pre_handle_cmd(str));

    return true;
end

function us_db.update(dbc, sync)
    assert(dbc, "db cmd must be not null");
    local code = dbdefine.ENUM_DBRESULT_ERRCODE.SUCCESS;
    local derr;
    local dmsg;

    local columns = dbc.m_columns;
    local values = dbc.m_values;
    dbc.m_columns = {};
    dbc.m_values = {};
    local srt = us_db.select(dbc);
    -- log.dump(srt);
    if dbdefine.ENUM_DBRESULT_ERRCODE.SUCCESS ~= srt.m_code then
        code = srt.m_code;
        derr = srt.m_derr;
        dmsg = srt.m_dmsg;
        log.printf("select error code ==> %s, msg ==> %s", code, dmsg);
        return dbstruct.create_result(code, derr, dmsg);
    end
    dbc.m_columns = columns;
    dbc.m_values = values;

    -- 更新redis
    if false == _update_redis(dbc) then
        code = dbdefine.ENUM_DBRESULT_ERRCODE.RDFAILED;
        return dbstruct.create_result(code, derr, dmsg);
    elseif true ~= sync then
        return dbstruct.create_result(code, derr, dmsg);
    end

    -- 更新mysql
    local urt = _update_mysql(dbc);
    log.dump(urt);
    return dbstruct.create_result(code, derr, dmsg);
end
------------------------------------------------------------------

-- 同步数据到mysql
local function _sync_record(rdkey)
    local arr = string.split(rdkey, ":");
    local tbname = assert(arr[1]);
    local tbpkv = assert(arr[2]);

    local dbc = dbstruct.create_cmd(tbname);
    dbc.m_pk = tbpkv;

    local rt = _select_redis(dbc);
    -- log.dump(rt);
    if nil == rt or "true" == rt._del then
        return;
    end

    dbc.m_columns, dbc.m_values = _get_tab_keys_tab_values(rt);
    _update_mysql(dbc);
end


-- 同步列表到mysql
local function _sync_list(ul)
    assert(type(ul)=="table", "ul must be a table");
    for k, rdkey in pairs(ul) do
        -- log.print("===>sync");
        -- log.print(k, rdkey);
        _sync_record(rdkey);
    end
end


-- 同步线程
local function _sync_impl()
    while true do
        local ul = skynet.call(redis_pool, "lua", "zrange", nil, ulkey, "0", "-1");
        _sync_list(ul)
        skynet.sleep(500);
    end
end

function us_db.start()
    mysql_pool = skynet.uniqueservice("us_mysql");
    skynet.call(mysql_pool, "lua", "start");

    redis_pool = skynet.uniqueservice("us_redis");
    skynet.call(redis_pool, "lua", "start");

    us_dbsync = skynet.uniqueservice("us_dbsync");
    skynet.call(us_dbsync, "lua", "start");

    _load_data_to_redis();

    skynet.fork(_sync_impl);
end

function us_db.init()
    skynet.dispatch("lua", function(session, source, cmd, ...)
        local f = assert(us_db[cmd], "us_db cmd=>"..cmd.." not found");
        skynet.retpack(f(...));
    end)
    skynet.register(SERVICE_NAME);
end

skynet.start(us_db.init);

